package com.wsjj.yjh;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class KeyedListStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.socketTextStream("1.94.41.70",7777)
                        .map(value -> {
                            String[] split = value.split(",");
                            return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                        })
                                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                                return element.getTs();
                                            }
                                        }))
                                        .keyBy(WaterSensor::getId)
                                                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                                                    ListState<Integer> vcList ;

                                                    @Override
                                                    public void open(Configuration parameters) throws Exception {
                                                         vcList = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("vcList", Types.INT));
                                                    }

                                                    @Override
                                                    public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                                                        vcList.add(value.getVc());
                                                        List<Integer> List = new ArrayList<>();
                                                        for (Integer vc : vcList.get()) {
                                                            List.add(vc);

                                                        }
                                                        List.sort(new Comparator<Integer>() {
                                                            @Override
                                                            public int compare(Integer o1, Integer o2) {
                                                                return o2 - o1;
                                                            }
                                                        });

                                                        if(List.size() > 3){
                                                            for (int i = 3; i < List.size(); i++) {
                                                                List.remove(i);
                                                            }
                                                        }
                                                        vcList.update(List);


                                                        out.collect("传感器id为" + value.getId() + ",最大的3个水位值=" + List.toString());
                                                    }
                                                })
                                                        .print();


        env.execute();
    }
}
